package com.itheima.order.service.impl;

import com.alibaba.fastjson.JSON;
import com.itheima.order.entity.OrderEntity;
import com.itheima.order.mapper.OrderMapper;
import com.itheima.order.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Service
@Slf4j
public class OrderServiceImpl implements OrderService {

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    private boolean flag;

    @Override
    public boolean addOrder(OrderEntity orderEntity) {

        try {
            orderEntity.setIsDel(0);
            int insertResult = orderMapper.insert(orderEntity);

            if (insertResult >= 1) {
                //todo 基础同步消息发送消息
                //rocketMQTemplate.convertAndSend("order-topic", JSON.toJSONString(orderEntity));

                //todo 同步发送(必须要知道 broker的返回结果,SendResult)
                //SendResult sendResult = rocketMQTemplate.syncSend("order-topic", JSON.toJSONString(orderEntity));

                /*SendStatus sendStatus = sendResult.getSendStatus();
                //SEND_OK --> 0
                if (!(0 == sendStatus.ordinal())){
                    log.error("消息发送失败");
                    return flag;
                }*/

                //todo 异步发送
                /*rocketMQTemplate.asyncSend("order-topic", JSON.toJSONString(orderEntity), new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println("发送成功");
                    }

                    @Override
                    public void onException(Throwable e) {
                        System.out.println("发送失败");
                        System.out.println(e.getMessage());
                    }
                });*/

                //todo 单向发送
                //rocketMQTemplate.sendOneWay("order-topic", JSON.toJSONString(orderEntity));

                //todo 发送消息时,携带自定义标签stock
                //SendResult sendResult = rocketMQTemplate.syncSend("order-topic:stock", JSON.toJSONString(orderEntity));

                //todo 发送延时消息: 向库存发送,延迟10S
                //封装消息对象
                //Message<String> message = new GenericMessage<>(JSON.toJSONString(orderEntity));
                /**
                 * destination: 目的地
                 * message: 消息体
                 * timeout: 消息发送的超时时间
                 * delayLevel: 消息的延迟等级
                 */
                //SendResult sendResult = rocketMQTemplate.syncSend("order-topic:stock", message, 5000, 4);

                //todo 局部有序
                //SendResult sendResult = null;
                //for (int i = 0; i < 10; i++) {
                /**
                 * destination：消息目的地
                 * payload：消息内容
                 * hashKey：该值用于判断当前消息会进入哪个队列
                 *          计算公式：传入值 % 队列总数=队列id，如 6%4=2，值进入id为2的队列
                 */
                //sendResult = rocketMQTemplate.syncSendOrderly("order-topic:message", "message : "+i,"6");
                //}

                //todo 实现消息幂等性
                Message message = new Message();
                //设置消息内容 setBody 需要传输字节数组
                message.setBody(JSON.toJSONString(orderEntity).getBytes(StandardCharsets.UTF_8));
                //设置消息的唯一表示 UUID
                message.setKeys(UUID.randomUUID().toString());
                SendResult sendResult = null;
                for (int i = 0; i < 5; i++) {

                    sendResult = rocketMQTemplate.syncSend("order-topic:stock", message);
                }
                SendStatus sendStatus = sendResult.getSendStatus();
                //SEND_OK --> 0
                if (!(0 == sendStatus.ordinal())) {
                    log.error("消息发送失败");
                    return flag;
                }

                flag = true;
                return flag;
            }
        } catch (Exception e) {
            log.info(e.getMessage());
        }

        return flag;
    }

    //todo 分布式事务
    @Override
    @Transactional //事务注解
    public boolean delOrder(Integer id) {

        //根据id查询订单
        OrderEntity orderEntity = orderMapper.selectById(id);

       /* (错误演示)
        //修改订单中的删除状态is_del
        orderEntity.setIsDel(1);
        int result = orderMapper.updateById(orderEntity);

        //通知库存,根据订单中的商品数据,恢复库存数量
        rocketMQTemplate.syncSend("order-del-topic:stock",JSON.toJSONString(orderEntity));

        //模拟异常
        //(订单服务is_del没有修改成功,库存服务却恢复了库存)
        //一个功能经过两个服务,所以库存服务的@Transactional 不能控制到订单服务的@Transactional
        //int i = 1 / 0;

        if (result == 1){
            return true;
        }
        */


        //生产者发送事务消息
        org.springframework.messaging.Message<?> message = MessageBuilder.withPayload(JSON.toJSONString(orderEntity)).build();

        rocketMQTemplate.sendMessageInTransaction("order-del-topic:stock",message,null);

        return false;
    }
}
